package com.lingjtx.common.thread.pooling;

import com.lingjtx.common.core.handle.ObjectPool;
import com.lingjtx.common.core.handle.PooledObject;
import com.lingjtx.common.thread.config.ExecutorConfig;
import com.lingjtx.common.thread.core.PoolNameConst;
import com.lingjtx.common.thread.core.ThreadPoolManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Function;

public class ObjectPoolTemplate<T> implements ObjectPool<T>, InitializingBean, DisposableBean {
    private static final Logger log = LoggerFactory.getLogger(ObjectPoolTemplate.class);
    private final BlockingQueue<T> pool; // 池的存储容器，使用阻塞队列来管理对象
    private final ObjectFactory<T> factory; // 工厂，用于创建、验证和销毁对象
    private final AtomicInteger currentSize = new AtomicInteger(0); // 当前池中对象的数量
    private final AtomicInteger currentInUse = new AtomicInteger(0); // 当前正在使用的对象数量
    private final Map<T, Long> objectLastUsedTimes = new ConcurrentHashMap<>(); // 记录每个对象最后一次使用的时间
    private final AtomicInteger objectCreatedCount = new AtomicInteger(0); // 已创建的对象总数
    private final AtomicInteger objectDestroyedCount = new AtomicInteger(0); // 已销毁的对象总数

    // 可选回调钩子
    private Runnable refreshCallback;
    private final ReentrantLock lock = new ReentrantLock();
    private final String poolName;

    // 配置项
    private long maxWaitMillis = 5000; // 获取连接时的最大等待时间（毫秒）
    private long cleanupIntervalMillis = 120000; // 清理周期，单位：毫秒
    private long expansionThresholdMillis = 3000; // 当等待时间超过该值时，触发池扩展
    private long idleTimeoutMillis = 180000; // 空闲连接的最大超时，超过该时间的连接将被销毁
    private int maxSize = 10; // 连接池的最大值
    private int initialSize = 5;

    private final ThreadPoolManager threadPoolManager; // 线程池管理器，用于定时清理任务
    private ScheduledFuture<?> cleanupFuture; // 清理任务的定时任务

    private final Map<String, Object> resourceMonitorMap = new ConcurrentHashMap<>(); // 资源监控信息

    // 构造方法，初始化连接池并填充初始连接
    public ObjectPoolTemplate(String poolName, ObjectFactory<T> factory, ThreadPoolManager threadPoolManager) {
        this.poolName = poolName;
        this.factory = factory;
        this.threadPoolManager = threadPoolManager;
        pool = new LinkedBlockingQueue<>(maxSize);
        // 填充初始连接池
        for (int i = 0; i < initialSize; i++) {
            T newObj = factory.create(); // 创建新对象
            pool.offer(newObj); // 将新对象放入池中
            objectCreatedCount.incrementAndGet(); // 记录创建的对象数量
            currentSize.incrementAndGet(); // 更新池的当前大小
        }
        log.info("正在初始化对象池 [{}] 初始化大小 = {}, 最大容量 = {}", poolName, initialSize, maxSize);
    }

    public void refreshConfig(ExecutorConfig.ObjectPoolConfig config) {
        lock.lock();
        try {
            this.initialSize = config.getInitialSize();
            this.maxSize = config.getMaxSize();
            this.cleanupIntervalMillis = config.getCleanupIntervalMillis();
            this.expansionThresholdMillis = config.getExpansionThresholdMillis();
            this.idleTimeoutMillis = config.getIdleTimeoutMillis();
            this.maxWaitMillis = config.getMaxWaitMillis();

            if (refreshCallback != null) {
                refreshCallback.run();
            }
        } finally {
            lock.unlock();
        }
    }

    public void setRefreshCallback(Runnable refreshCallback) {
        this.refreshCallback = refreshCallback;
    }

    // 配置项的 setter 方法
    public void setMaxWaitMillis(long maxWaitMillis) {
        this.maxWaitMillis = maxWaitMillis;
    }

    public void setCleanupIntervalMillis(long cleanupIntervalMillis) {
        this.cleanupIntervalMillis = cleanupIntervalMillis;
    }

    public void setExpansionThresholdMillis(long expansionThresholdMillis) {
        this.expansionThresholdMillis = expansionThresholdMillis;
    }

    public void setIdleTimeoutMillis(long idleTimeoutMillis) {
        this.idleTimeoutMillis = idleTimeoutMillis;
    }

    public void setMaxSize(int maxSize) {
        this.maxSize = maxSize;
    }

    public void setInitialSize(int initialSize) {
        this.initialSize = initialSize;
    }

    // 获取池中的一个对象（连接），如果没有可用的连接，则会等待或扩展池
    @Override
    public T acquire() {
        T obj = pool.poll(); // 从池中获取一个对象，如果有空闲的对象则直接返回
        if (obj != null && factory.validate(obj)) {
            currentInUse.incrementAndGet(); // 更新正在使用的连接数
            objectLastUsedTimes.put(obj, System.currentTimeMillis()); // 记录最后使用时间
            return obj;
        }

        // 如果当前池大小小于最大池大小，可以创建新的对象
        if (currentSize.get() < maxSize) {
            lock.lock();
            try {
                if (currentSize.get() < maxSize) {
                    T newObj = factory.create(); // 创建新的对象
                    currentSize.incrementAndGet();
                    objectCreatedCount.incrementAndGet(); // 更新已创建对象的计数
                    currentInUse.incrementAndGet(); // 更新正在使用的连接数
                    objectLastUsedTimes.put(newObj, System.currentTimeMillis()); // 记录最后使用时间
                    pool.offer(newObj); // 将新创建的对象放入池中
                    return newObj;
                }
            } finally {
                lock.unlock();
            }
        }

        // 如果无法立即获取连接，则等待指定的最大时间
        long startWaitTime = System.currentTimeMillis();
        try {
            T waitObj = pool.poll(maxWaitMillis, TimeUnit.MILLISECONDS); // 等待池中释放的连接
            if (waitObj != null && factory.validate(waitObj)) {
                currentInUse.incrementAndGet(); // 更新正在使用的连接数
                objectLastUsedTimes.put(waitObj, System.currentTimeMillis()); // 更新最后使用时间
                return waitObj;
            }

            // 最后一搏：快速再尝试从池中获取一次，避免刚释放错过
            waitObj = pool.poll();
            if (waitObj != null && factory.validate(waitObj)) {
                currentInUse.incrementAndGet();
                objectLastUsedTimes.put(waitObj, System.currentTimeMillis());
                return waitObj;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 中断异常处理
        }

        // 如果等待时间超过阈值，则尝试扩展池
        if (System.currentTimeMillis() - startWaitTime > expansionThresholdMillis) {
            expandPool();
        }

        throw new RuntimeException("无法在指定时间内获取可用连接");
    }

    // 释放一个连接回池
    @Override
    public void release(T obj) {
        if (factory.validate(obj)) {
            // 如果连接仍然有效，将其返回池中
            if (!pool.offer(obj)) {
                factory.destroy(obj); // 如果池满了，销毁该连接
                objectDestroyedCount.incrementAndGet(); // 更新销毁对象计数
                currentSize.decrementAndGet(); // 更新池的大小
                tryRefill(); // 尝试填充连接池
            }
            currentInUse.decrementAndGet(); // 更新使用中的连接数
        } else {
            // 如果连接无效，销毁它并更新相关计数
            factory.destroy(obj);
            objectDestroyedCount.incrementAndGet();
            currentSize.decrementAndGet();
            currentInUse.decrementAndGet();
            tryRefill(); // 尝试填充连接池
        }
    }

    /**
     * 获取池中的一个对象（连接）,操作完之后自动释放对象
     * 通过 try-with-resources 语法糖获取对象
     */
    @Override
    public PooledObject<T> autoAcquire() {
        T obj = acquire();
        return new PooledObject<>(obj, this); // 返回自动释放包装器
    }

    @Override
    public <R> R execute(Function<T, R> action) {
        try (PooledObject<T> pooled = autoAcquire()) {
            return action.apply(pooled.get());
        }
    }

    @Override
    public void consume(Consumer<T> consumer) {
        try (PooledObject<T> pooled = autoAcquire()) {
            consumer.accept(pooled.get());
        }
    }

    @Override
    public void asyncConsume(Consumer<T> consumer) {
        async(() -> {
            try (PooledObject<T> pooled = autoAcquire()) {
                consumer.accept(pooled.get());
            }
        });
    }


    @Override
    public <R> CompletableFuture<R> asyncExecute(Function<T, R> action) {
        return CompletableFuture.supplyAsync(() -> {
            try (PooledObject<T> pooled = autoAcquire()) {
                return action.apply(pooled.get());
            }
        }, this::async);
    }

    @Override
    public void async(Runnable task) {
        threadPoolManager.execute(PoolNameConst.POOLING_OBJECT, task);
    }


    // 销毁方法，在销毁池时调用
    @Override
    public void destroy() {
        shutdown();
        factory.shutdown(); // 调用工厂的销毁方法
        log.info("对象 [{}] 池完成销毁", poolName);
    }

    // 初始化方法，设置定时清理任务和资源监控任务
    @Override
    public void afterPropertiesSet() {
        ScheduledExecutorService scheduler = threadPoolManager.getOrCreateScheduled(PoolNameConst.SCHEDULED_CLEANER);
        this.cleanupFuture = scheduler.scheduleAtFixedRate(
                this::clearInvalidObjects,
                cleanupIntervalMillis,
                cleanupIntervalMillis,
                TimeUnit.MILLISECONDS
        );
        startResourceMonitoring(); // 启动资源监控任务
    }

    // 获取资源监控信息
    @Override
    public Map<String, Object> getResourceMonitorMap() {
        return Collections.unmodifiableMap(resourceMonitorMap); // 返回不可修改的资源监控Map
    }

    // 关闭连接池并销毁所有对象
    @Override
    public void shutdown() {
        if (cleanupFuture != null && !cleanupFuture.isCancelled()) {
            cleanupFuture.cancel(true); // 取消清理任务
        }
        for (T obj : pool) {
            factory.destroy(obj); // 销毁池中的每个对象
        }
        pool.clear(); // 清空池
        threadPoolManager.shutdownAll(PoolNameConst.POOLING_OBJECT, PoolNameConst.SCHEDULED_CLEANER, PoolNameConst.SCHEDULED_MONITOR);
    }


    // 清理无效对象：检查连接池中的每个对象，销毁无效的对象
    private void clearInvalidObjects() {
        int minimumRetained = 3; // 至少保留3个对象
        int scannedCount = 0;
        int maxScan = Math.min(maxSize, pool.size()); // 每次最多扫描 maxSize 个（可调节）
        List<T> removed = new ArrayList<>();

        Iterator<T> iterator = pool.iterator();
        while (iterator.hasNext() && scannedCount < maxScan) {
            T obj = iterator.next();
            scannedCount++;

            if (pool.size() - removed.size() <= minimumRetained) {
                break; // 保留至少 minimumRetained 个对象
            }

            long lastUsedTime = objectLastUsedTimes.getOrDefault(obj, 0L);
            boolean expired = System.currentTimeMillis() - lastUsedTime > idleTimeoutMillis;
            // 判断连接是否无效，或者是否空闲时间过长
            if (!factory.validate(obj) || expired) {
                removed.add(obj);// 记录待删除的对象
                currentSize.decrementAndGet(); // 更新池的大小
                objectDestroyedCount.incrementAndGet(); // 更新销毁对象计数
            }
        }
        removed.forEach(factory::destroy); // 销毁所有无效对象
        pool.removeAll(removed); // 从池中移除无效对象
    }

    // 扩展池的大小，当池大小小于最大值时，增加池中的连接数
    private void expandPool() {
        async(() -> {
            if (currentSize.get() < maxSize) {
                int newSize = Math.min(maxSize, currentSize.get() + (maxSize / 4)); // 增加池大小的四分之一
                if (newSize > maxSize) newSize = maxSize; // 确保不超过最大池大小
                this.maxSize = newSize;

                // 根据新的池大小创建新的对象
                while (currentSize.get() < maxSize) {
                    T newObj = factory.create();
                    pool.offer(newObj); // 放入池中
                    objectCreatedCount.incrementAndGet(); // 更新创建的对象计数
                    currentSize.incrementAndGet(); // 更新池的大小
                    objectLastUsedTimes.put(newObj, System.currentTimeMillis()); // 记录最后使用时间
                }
            }
        });
    }

    // 启动资源监控任务
    private void startResourceMonitoring() {
        ScheduledExecutorService scheduler = threadPoolManager.getOrCreateScheduled(PoolNameConst.SCHEDULED_MONITOR);
        scheduler.scheduleAtFixedRate(this::monitorResourceUsage, 0, 30, TimeUnit.SECONDS); // 每30秒更新一次监控信息
    }

    // 监控资源的使用情况，并更新资源监控信息
    private void monitorResourceUsage() {
        resourceMonitorMap.put("当前池大小", pool.size()); // 更新当前池大小
        resourceMonitorMap.put("当前使用中的连接数", currentInUse.get()); // 更新当前使用中的连接数
        resourceMonitorMap.put("当前总连接数", currentSize.get()); // 更新总连接数
        resourceMonitorMap.put("最大池大小", maxSize); // 最大池大小
        resourceMonitorMap.put("已创建对象数", objectCreatedCount.get()); // 已创建的对象数
        resourceMonitorMap.put("已销毁对象数", objectDestroyedCount.get()); // 已销毁的对象数
        resourceMonitorMap.put("平均空闲连接时间", calculateAverageIdleTime() + "ms"); // 计算平均空闲时间
    }

    // 计算平均空闲时间
    private long calculateAverageIdleTime() {
        long totalIdleTime = 0;
        long currentTime = System.currentTimeMillis();
        int count = 0;
        for (Map.Entry<T, Long> entry : objectLastUsedTimes.entrySet()) {
            totalIdleTime += (currentTime - entry.getValue()); // 计算空闲时间
            count++;
        }
        return count > 0 ? totalIdleTime / count : 0; // 返回平均空闲时间
    }

    // 尝试为连接池添加新对象
    private void tryRefill() {
        if (currentSize.get() < maxSize) {
            if (currentSize.incrementAndGet() <= maxSize) {
                T newObj = factory.create();
                if (factory.validate(newObj)) {
                    pool.offer(newObj); // 将新对象放入池中
                } else {
                    currentSize.decrementAndGet(); // 如果创建的对象无效，则减少池的大小
                }
            } else {
                currentSize.decrementAndGet(); // 如果池已满，则恢复池的大小
            }
        }
    }
}